/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.service.impl;

import cn.sunny360.core.common.PhResult;
import cn.sunny360.core.service.MessageService;
import cn.sunny360.core.utils.Container;
import cn.sunny360.core.utils.ZipStrUtil;
import cn.sunny360.core.wrapper.CacheWrapper;
import cn.sunny360.core.wrapper.DBWrapper;
import com.sohu.idcenter.IdWorker;
import io.vertx.core.Promise;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.core.json.JsonObject;
import io.vertx.sqlclient.Tuple;

import java.util.List;
import java.util.stream.Collectors;

/**
 * implementation of {@link MessageService}.
 *
 * @author sublun
 */
public class MessageServiceImpl implements MessageService {

	private static final Logger logger = LoggerFactory.getLogger(MessageServiceImpl.class);

	private IdWorker idWorker;
	private DBWrapper dbWrapper;
	private CacheWrapper cacheWrapper;

	public MessageServiceImpl() {
		idWorker = Container.getComponent(IdWorker.class);
		dbWrapper = Container.getComponent(DBWrapper.class);
		cacheWrapper = Container.getComponent(CacheWrapper.class);
	}


	@Override
	public MessageServiceImpl saveMessage(JsonObject message, Promise<Long> resultHandler) {
		// 生成message的version
		cacheWrapper.genVersion(h->{
			Long version = h.result();
			message.put("version", version);
			//保存message
			Promise<Integer> promise = dbWrapper.insert(
					"insert into ph_message(\n" +
							"  message_id,\n" +
							"  msg_from,\n" +
							"  msg_to,\n" +
							"  msg_type,\n" +
							"  msg_body,\n" +
							"  msg_group,\n" +
							"  is_delivery,\n" +
							"  sendtime\n" +
							") values (\n" +
							"  ?,?,?,?,?,?,?,?\n" +
							")", Tuple.of(
							version,
							Long.parseLong(message.getString("from")),
							Long.parseLong(message.getString("to")),
							message.getString("type"),
							ZipStrUtil.zipBase64(message.toString()),
							message.getString("group"),
							0,
							message.getString("time")
					));
			promise.future()
					.onFailure(e->{
						// TODO 如果主键冲突需要重新取主键
						resultHandler.fail(e);
					})
					.onSuccess(result->resultHandler.complete(version));
		});

		return this;
	}

	@Override
	public MessageServiceImpl getMessageList(Long userId, Long lastVersion,
                                             Promise<List<String>> resultHandler) {
		Promise<List<JsonObject>> promise = dbWrapper
				.findList("select msg_body from ph_message where msg_to=? and message_id > ? order by message_id",
						Tuple.of(userId, lastVersion));
		promise.future()
				.onFailure(resultHandler::fail)
				.onSuccess(result->{
					List<String> msgList = result.stream()
							.map(e -> e.getString("msg_body"))
							.collect(Collectors.toList());
					resultHandler.complete(msgList);
				});
		return this;
	}

	@Override
	public MessageService updateMessageStatus(Long version, Promise<PhResult> handler) {
		dbWrapper.update("update ph_message set is_delivery = 1 where message_id=?",
				Tuple.of(version))
				.future()
				.onSuccess(h->handler.complete(PhResult.ok()))
				.onFailure(handler::fail);
		return this;
	}

	@Override
	public MessageService getGroupMessageList(Long groupId, Long lastVersion, Promise<List<String>> resultHandler) {
		dbWrapper.findList("select msg_body from ph_message where msg_group = ? and message_id > ?",
				Tuple.of(groupId, lastVersion))
				.future()
				.onSuccess(list->{
					List<String> msgList = list.stream()
							.map(e->e.getString("msg_body"))
							.collect(Collectors.toList());
					resultHandler.complete(msgList);
				}).onFailure(resultHandler::fail);
		return this;
	}

	@Override
	public MessageService getHistoryMessageList(long userId, Long lastVersion, long offset, Promise<List<String>> resultHandler) {
		Promise<List<JsonObject>> promise = dbWrapper
				.findList("select * from (\n" +
								"    select * from (\n" +
								"        SELECT message_id, msg_body \n" +
								"                    FROM `ph_message` \n" +
								"                 WHERE msg_from = ?\n" +
								"        union\n" +
								"        SELECT message_id, msg_body \n" +
								"                    FROM `ph_message` \n" +
								"                 WHERE msg_to = ? \n" +
								"                     AND message_id <= ?\n" +
								"        union\n" +
								"        SELECT t1.message_id, t1.msg_body \n" +
								"                    FROM `ph_message` t1 join ph_user_group t2 \n" +
								"                        on t1.msg_to = t2.group_id\n" +
								"                     and t1.message_id <= t2.group_version \n" +
								"                WHERE t2.user_id = ?\n" +
								"    ) a\n" +
								"    ORDER BY a.message_id DESC \n" +
								"    LIMIT ?\n" +
								") b \n" +
								"ORDER BY b.message_id ",
						Tuple.of(userId, userId, lastVersion, userId, offset));
		promise.future()
				.onFailure(resultHandler::fail)
				.onSuccess(result->{
					List<String> msgList = result.stream()
							.map(e -> e.getString("msg_body"))
							.collect(Collectors.toList());
					resultHandler.complete(msgList);
				});
		return this;
	}

}